package queue

import (
	"errors"
	"github.com/upmio/horus-agent/model"
	"github.com/upmio/horus-polymerizer/log"
	"sync"
)

//EventQueue is exported
type EventQueue struct {
	curlen     int
	maxlen     int
	datas      chan *model.EventMetric
	lock       *sync.Mutex
	name       string
	discardNum int
}

var (
	globaleventQue      *EventQueue
	globalKafkaEventQue *EventQueue
)

//NewEventQueue is exported
func NewEventQueue(qlen, discardNum int) (*EventQueue, error) {
	if qlen < discardNum+100 {
		return nil, errors.New("the qlen must bigger than discardNum+200")
	}

	return &EventQueue{
		curlen:     0,
		maxlen:     qlen,
		datas:      make(chan *model.EventMetric, qlen),
		lock:       new(sync.Mutex),
		name:       "defaultEvent",
		discardNum: discardNum,
	}, nil
}

//GetDefaultKafkaEventQue is exported
func GetDefaultKafkaEventQue() *EventQueue {

	once.Do(func() {
		if globalKafkaEventQue == nil {
			globalKafkaEventQue = &EventQueue{
				curlen:     0,
				maxlen:     20000,
				datas:      make(chan *model.EventMetric, 2000),
				lock:       new(sync.Mutex),
				name:       "defaultKafkaEvent",
				discardNum: 500,
			}
		}
	})

	return globalKafkaEventQue

}

//InitEventQueue is exported
func InitEventQueue(qlen, discardNum int) error {
	var err error
	globaleventQue, err = NewEventQueue(qlen, discardNum)
	return err
}

//PutEvent is exported
func PutEvent(data *model.EventMetric) error {
	return globaleventQue.PutEvent(data)

}

//GetEventSlice is exported
func GetEventSlice() ([]*model.EventMetric, error) {
	return globaleventQue.GetEventSlice()
}

//GetEventCurLen is exported
func GetEventCurLen() int {
	return globaleventQue.GetCurLen()
}

//GetEventMaxLen is exported
func GetEventMaxLen() int {
	return globaleventQue.GetMaxLen()
}

//GetCurLen is exported
func (q *EventQueue) GetCurLen() int {
	return q.curlen
}

//GetMaxLen is exported
func (q *EventQueue) GetMaxLen() int {
	return q.maxlen
}

//GetDiscardNum is exported
func (q *EventQueue) GetDiscardNum() int {
	return q.discardNum
}

func (q *EventQueue) subCurlen(num int) {
	q.lock.Lock()
	q.curlen -= num
	q.lock.Unlock()
}

func (q *EventQueue) addCurlen(num int) {
	q.lock.Lock()
	q.curlen += num
	q.lock.Unlock()
}

func (q *EventQueue) delEventMetric(num int) {
	for i := 0; i < num; i++ {
		//write to a file?
		_, _ = q.GetEvent()
	}
	log.Warn("event queue  full,del some.", "now len:", q.GetCurLen())
}

//PutEvent is exported
//放数据到该队列中
func (q *EventQueue) PutEvent(data *model.EventMetric) error {

	q.addCurlen(1)
	if q.curlen >= q.maxlen {
		q.delEventMetric(q.discardNum)
	}
	q.datas <- data

	log.Debug("evenque curlen::::: ", q.GetCurLen(), "   evenque maxlen:::", q.GetMaxLen())
	return nil
}

//GetEvent is exported
//it will block，which  I need.
func (q *EventQueue) GetEvent() (*model.EventMetric, error) {

	//if have no data ,it will block, which  I need.
	data := <-q.datas

	//should not happen
	if data == nil {
		log.Error("get event data from queue err.")
		return nil, errors.New("get error")
	}
	//it doesnt matter that after get data..
	q.subCurlen(1)
	return data, nil
}

//GetEventSlice is exported
func (q *EventQueue) GetEventSlice() ([]*model.EventMetric, error) {
	curlen := q.GetCurLen()
	// events := make([]*model.EventMetric, 0, 8)
	events := []*model.EventMetric{}

	if curlen >= 8 {
		curlen = 8
	} else if curlen == 0 {
		curlen = 1
	}

	for i := 0; i < curlen; i++ {
		event, _ := q.GetEvent()
		events = append(events, event)
	}

	return events, nil
}
